package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo04EventTime {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> lineDS = env.socketTextStream("master", 8888);

        /*
         * 2024-12-30 15:14:51 ====>  1735542891000 ms
         *
         * a,1735542891000
         * a,1735542892000
         * a,1735542893000
         * a,1735542894000
         * a,1735542895000
         * a,1735542896000
         * a,1735542897000
         * a,1735542898000
         * a,1735542899000
         * a,1735542900000
         * a,1735542901000
         * a,1735542906000
         */
        // 整理数据
        DataStream<Tuple2<String, Long>> wordsTSDS = lineDS.map(line -> {
            String[] split = line.split(",");
            String word = split[0];
            Long ts = Long.parseLong(split[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        // 告诉Flink，第二列是时间戳
        DataStream<Tuple2<String, Long>> assignDS = wordsTSDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                // 连续递增水位线策略，不考虑数据的乱序，如果数据乱序到达（在窗口结束之后到达）则会被丢弃
                                // 时间戳在Flink中只会被认为是一直增加的，以到达数据中最大的时间戳为当前整个Job的时间戳
                                // 连续递增水位线策略中：水位线就等于最大的时间戳
//                                .<Tuple2<String, Long>>forMonotonousTimestamps()
                                // 固定等待时间的水位策略（水位线前移策略），可以容忍5s的乱序到达问题
                                // 导致整个Job会增大5s的延迟
                                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                /*
                                 * 指定哪一列是时间戳
                                 * withTimestampAssigner中需要传入一个Lambada
                                 * Lambada表达式接收两个参数：t2,ts
                                 * t2:DataStream中的每一条数据
                                 * ts：方法自带的，必须要加上，不需要处理
                                 * 最终需要从DS中一条数据中去除时间戳列进行返回
                                 */
                                .withTimestampAssigner((t2, ts) -> t2.f1)


                );

        // 如果要做基于时间的统计，那么必须要通过时间窗口操作完成
        assignDS
                .map(t2 -> Tuple2.of(t2.f0, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .window(EventTimeSessionWindows.withGap(Time.seconds(5)))
                .sum(1)
                .print();


        env.execute();


    }
}
